//package com.k2data.platform.service.adapters.fixedformat.topology;
//
//import backtype.storm.Config;
//import backtype.storm.LocalCluster;
//import backtype.storm.StormSubmitter;
//import backtype.storm.generated.AlreadyAliveException;
//import backtype.storm.generated.AuthorizationException;
//import backtype.storm.generated.InvalidTopologyException;
//import backtype.storm.spout.SchemeAsMultiScheme;
//import backtype.storm.topology.TopologyBuilder;
//
//import com.k2data.platform.ddm.share.common.Configuration;
//import com.k2data.platform.ddm.share.common.Constants;
//import com.k2data.platform.ddm.share.common.ParamNames;
//import com.k2data.platform.ddm.share.common.storm.KafkaKeyValueScheme;
//import com.k2data.platform.ddm.share.common.zookeeper.ZookeeperService;
//import com.k2data.platform.service.adapters.fixedformat.bolt.RedisAvroBolt;
//import org.apache.commons.configuration.ConfigurationException;
//import org.apache.commons.lang.StringUtils;
//import org.apache.storm.shade.org.apache.zookeeper.client.ConnectStringParser;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import storm.kafka.*;
//
//import java.io.FileInputStream;
//import java.io.IOException;
//import java.net.InetSocketAddress;
//import java.util.ArrayList;
//import java.util.List;
//import java.util.Map;
//import java.util.Properties;
//
///**
// * Created by luoqifei on 17-5-3.
// */
//public class RedisTopology {
//    private final static Logger logger = LoggerFactory.getLogger(RedisTopology.class);
//    private final static String ID_KAFKA_SPOUT = "redis-read-kafka-spout";
//    private final static String ID_REDIS_BOLT = "avro-save2redis-bolt";
//    private Map<String, String> env;
//    private static Configuration configuration;
//    private final static String K2_ADAPTER_MODE = "K2_ADAPTER_MODE";
//
//    // necessary configuration item
//    private static String brokerList;
//    private static String zkUrl;
//    private static String nimbus;
//
//    public RedisTopology() {
//        env = System.getenv();
//        configuration = getConfiguration();
//    }
//    public Configuration getConfiguration() {
//        Configuration configuration = new Configuration();
//
//        if (env.containsKey(K2_ADAPTER_MODE) && env.get(K2_ADAPTER_MODE).equals("local")) {
//            // load local env file
//            logger.info("Running in local mode ...");
//            Properties localEnv = new Properties();
//            try {
//                localEnv.load(new FileInputStream(getClass().getResource("/").getPath().toString() + "../../env.qa"));
//                for (Object key : localEnv.keySet()) {
//                    configuration.setProperty(String.valueOf(key).toLowerCase().replace('_', '.').trim(), String.valueOf(localEnv.get(key)).replaceAll("\"", ""));
//                }
//            } catch (IOException e) {
//                logger.error("Fail to load env.local file. Will exit with error.");
//                e.printStackTrace();
//                System.exit(1);
//            }
//            logger.info("Success to load env.local file: {}", configuration.toString());
//        } else {
//            logger.info("Running in remote mode.");
//            // set env variables into configuration. Like setting METADATA_BROKER_LIST to metadata.broker.list value.
//            for (String key : env.keySet()) {
//                configuration.setProperty(key.toLowerCase().replace('_', '.').trim(), env.get(key));
//            }
//            logger.info("Success to load env setting: {}", configuration.toString());
//        }
//
//        // verify necessary configuration
//        zkUrl = configuration.getString(ParamNames.ZOOKEEPER_URL);
//        if (StringUtils.isBlank(zkUrl)) {
//            logger.error("Not set env {}", ParamNames.ZOOKEEPER_URL.toUpperCase().replace('.', '_').trim());
//            System.exit(1);
//        }
//        brokerList = configuration.getString(ParamNames.METADATA_BROKER_LIST);
//        if (StringUtils.isBlank(brokerList)) {
//            logger.error("Not set env {}", ParamNames.METADATA_BROKER_LIST.toUpperCase().replace('.', '_').trim());
//            System.exit(1);
//        }
//        nimbus = configuration.getString(ParamNames.NIMBUS_HOST);
//        if (StringUtils.isBlank(nimbus)) {
//            logger.error("Not set env {}", ParamNames.NIMBUS_HOST.toUpperCase().replace('.', '_').trim());
//            System.exit(1);
//        }
//
//        return configuration;
//    }
//
//    public static void main(String[] args) {
//        Config config = new Config();
//        // create topology
//        RedisTopology topology = new RedisTopology();
//        TopologyBuilder builder = new TopologyBuilder();
//
//        builder.setSpout(ID_KAFKA_SPOUT, topology.getKafkaSpout(configuration),
//                configuration.getInt(ParamNames.ADAPTER_TOPOLOGY_SPOUT_PARALLELISM, Constants.ADAPTER_TOPOLOGY_SPOUT_PARALLELISM));
//        builder.setBolt(ID_REDIS_BOLT, topology.getRedisAvroBolt(configuration),
//                configuration.getInt(ParamNames.ADAPTER_TOPOLOGY_TRANSFORM_BOLT_PARALLELISM,
//                        Constants.ADAPTER_TOPOLOGY_TRANSFORM_BOLT_PARALLELISM)).noneGrouping(ID_KAFKA_SPOUT);
//
//
//        // set Storm properties
//        config.setNumWorkers(configuration.getInt(ParamNames.ADAPTER_TOPOLOGY_WORKER, Constants.ADAPTER_TOPOLOGY_WORKER));
//        config.setDebug(configuration.getBoolean(ParamNames.STORM_DEBUG, Constants.STORM_DEBUG));
//        config.put(Config.NIMBUS_HOST, nimbus);
//        config.setMaxSpoutPending(1000);
//
//        if (System.getenv().containsKey(K2_ADAPTER_MODE) && System.getenv().get(K2_ADAPTER_MODE).equals("local")) {
//            // submit topology locally
//            LocalCluster cluster = new LocalCluster();
//            cluster.submitTopology(RedisTopology.class.getSimpleName(), config, builder.createTopology());
//            while(true);
//        } else {
//            // submit topology remotely
//            try {
//                StormSubmitter.submitTopologyWithProgressBar(RedisTopology.class.getSimpleName(), config, builder.createTopology());
//            } catch (AlreadyAliveException e) {
//                e.printStackTrace();
//            } catch (InvalidTopologyException e) {
//                e.printStackTrace();
//            } catch (AuthorizationException e) {
//                e.printStackTrace();
//            }
//        }
//
//        System.exit(0);
//    }
//    public KafkaSpout getKafkaSpout(Configuration configuration) {
//        String topic = configuration.getString(ParamNames.KAFKA_TOPIC_NAME_MONITOR);
//        // topic="defaultUser-defaultSpace-raw";
//        String zkRoot = "/" + configuration.getString(ParamNames.ZOOKEEPER_NAMESPACE, Constants.ZOOKEEPER_NAMESPACE) + Constants.ZOOKEEPER_STORM_PATH;
//        String id = ID_KAFKA_SPOUT;
//        BrokerHosts brokerHosts = new ZkHosts(zkUrl);
//        SpoutConfig spoutConf = new SpoutConfig(brokerHosts, topic, zkRoot, id);
//        spoutConf.scheme = new KeyValueSchemeAsMultiScheme(new KafkaKeyValueScheme());
//        spoutConf.retryDelayMaxMs = 60000;
//        spoutConf.retryInitialDelayMs = 6000;
//        spoutConf.retryDelayMultiplier = 2;
//
//        //set the zkServers and zkPort for saving offset information into zookeeper "zkRoot/id"
//        ConnectStringParser parser = new ConnectStringParser(zkUrl);
//        List<InetSocketAddress> serverInetAddresses = parser.getServerAddresses();
//        List<String> serverAddresses = new ArrayList<String>(serverInetAddresses.size());
//        spoutConf.zkPort = serverInetAddresses.get(0).getPort();
//        for (InetSocketAddress address : serverInetAddresses) {
//            serverAddresses.add(address.getHostName());
//        }
//        spoutConf.zkServers = serverAddresses;
//
//        return new KafkaSpout(spoutConf);
//    }
//    public RedisAvroBolt getRedisAvroBolt(Configuration configuration) {
//        RedisAvroBolt bolt = null;
//        try {
//            bolt = new RedisAvroBolt(configuration.toBytes());
//        } catch (ConfigurationException e) {
//            logger.error("Fail to convert Configuration to byte[]!");
//            e.printStackTrace();
//            System.exit(1);
//        } catch (IOException e) {
//            logger.error("Fail to convert Configuration to byte[]!");
//            e.printStackTrace();
//            System.exit(1);
//        }
//        return bolt;
//    }
//}
